package samples

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.cancelChildren
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.TickerMode
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.channels.produce
import kotlinx.coroutines.channels.ticker
import kotlinx.coroutines.channels.toList
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.test.runTest
import kotlinx.coroutines.withTimeoutOrNull
import kotlinx.coroutines.yield
import org.junit.Test
import kotlin.system.measureTimeMillis

/**
 *
 * @author mph
 * @date 2025/1/13
 */
@OptIn(ExperimentalCoroutinesApi::class)
class ChannelTest {

    @Test
    fun test1() = runTest {
        val channel = Channel<Int>()
        launch {
            for (x in 1..5) channel.send(x * x)
            channel.close() // we're done sending 作为一个事件被send，保证前面的send事件可以被正常接收
        }
        // here we print received values using `for` loop (until the channel is closed)
        for (y in channel) println(y)
        println("Done!")
    }

    private fun CoroutineScope.produceSquares(): ReceiveChannel<Int> = produce {
        for (x in 1..5) send(x * x)
    }

    @Test
    fun test2() = runTest {
        val squares = produceSquares()
        squares.consumeEach { println(it) }
        println("Done!")
    }

    /**
     * 质数生成器
     * 两种方式：channel（优势：当使用Dispatchers.Default时利用更多CPU核心工作）和iterator
     * 注意，使用更多CPU核心不一定耗时更少，比如这里iterator方式耗时更少
     * channel pipeline适合处理异步操作，比如远程接口调用等，这些场景下iterator无法胜任
     */
    @Test
    fun test3() = runTest {
        val time = measureTimeMillis {
            runBlocking {
                var cur = numbersFrom(2) //得到一个2，3，4，5，6，7，8，9，10...流
                //第一遍，输出第一个send的元素：2，然后过滤掉能被2整除的数，留下3、5、7、9、11、13...
                //第二遍，输出第一个send的元素：3，然后过滤掉能被3整除的数，留下5、7、11、13、17、19、23、25、29、31、35、37、41、43、47、49...
                //第三遍，输出第一个send的元素：5，然后过滤掉能被5整除的数，留下7、11、13、17、19、23、29、31、37、41、43、47、49...
                //第四遍，输出第一个send的元素：7，然后过滤掉能被7整除的数，留下11、13、17、19、23、29、31、37、41、43、47...
                //一直循环10遍就会得到10个质数
                repeat(10) {
                    val prime = cur.receive()
                    println(prime)
                    cur = filter(cur, prime) //每次循环完，下次循环都会操作本次过滤后的结果流
                }
                coroutineContext.cancelChildren() // cancel all children to let main finish
            }
        }

        //iterator方式
        /* val time = measureTimeMillis {
             var numbersFromByIterator = numbersFromByIterator(2)
             repeat(10) {
                 val prime = numbersFromByIterator.next()
                 println(prime)
                 numbersFromByIterator = filterByIterator(numbersFromByIterator, prime)
             }
         }*/

        println("Time : $time")
    }

    private fun CoroutineScope.numbersFrom(start: Int) = produce<Int> {
        var x = start
        while (true) {
//            println("number: $x")
            send(x++) // infinite stream of integers from start
        }
    }

    private fun CoroutineScope.numbersFromByIterator(start: Int) = iterator {
        var x = start
        while (true) {
            yield(x++)
        }
    }

    private fun CoroutineScope.filter(numbers: ReceiveChannel<Int>, prime: Int) = produce<Int> {
        for (x in numbers) {
//            println("x: $x , prime: $prime")
            if (x % prime != 0) send(x)
        }
    }

    private fun CoroutineScope.filterByIterator(numbers: Iterator<Int>, prime: Int) = iterator<Int> {
        while (numbers.hasNext()) {
//            println("x: $x , prime: $prime")
            val x = numbers.next()
            if (x % prime != 0) yield(x)
        }
    }

    @Test
    fun test4() = runTest {
        val tickerChannel = ticker(mode = TickerMode.FIXED_DELAY, delayMillis = 200, initialDelayMillis = 0) // create a ticker channel
        var nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
        println("Initial element is available immediately: $nextElement") // no initial delay

        nextElement = withTimeoutOrNull(100) { tickerChannel.receive() } // all subsequent elements have 200ms delay
        println("Next element is not ready in 100 ms: $nextElement")

        nextElement = withTimeoutOrNull(120) { tickerChannel.receive() }
        println("Next element is ready in 200 ms: $nextElement")

        // Emulate large consumption delays
        println("Consumer pauses for 300ms")
        delay(300)
        // Next element is available immediately
        nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
        println("Next element is available immediately after large consumer delay: $nextElement")
        // Note that the pause between `receive` calls is taken into account and next element arrives faster
        nextElement = withTimeoutOrNull(120) { tickerChannel.receive() }
        println("Next element is ready in 100ms after consumer pause in 300ms: $nextElement")

        tickerChannel.cancel() // indicate that no more elements are needed
    }

}